package com.hph.flinkcdc;


import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * Flink CDC sql方式
 */
public class FlinkCDC_Sql {
    public static void main(String[] args) throws Exception {

        // 获取flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 开启checkpoint,实现断点续传
        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);
        // 设置 checkpoint 超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 设置 checkpoint 存储位置
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        // 设置最多可以同时进行多少个 checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 设置检查点模式（恰好一次与至少一次）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 使用Flink CDC SQL方式建表
        tableEnv.executeSql("CREATE TABLE t1 (\n" +
                "  id INT,\n" +
                "  name STRING,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = 'localhost',\n" +
                "  'port' = '3306',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'database-name' = 'test',\n" +
                "  'table-name' = 't1'\n" +
                ")");

        // 查询并打印
        Table table = tableEnv.sqlQuery("select * from t1");
        table.execute().print();
    }
}
